package org.groupg.project;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
import org.jgrapht.Graph;
import org.jgrapht.alg.cycle.CycleDetector;
import org.jgrapht.graph.DefaultDirectedGraph;
import org.jgrapht.graph.DefaultEdge;
import org.jgrapht.traverse.TopologicalOrderIterator;

import java.util.*;
import java.util.logging.Logger;

public class DisruptorWithJGraphTExample {

    static final Logger log = Logger.getLogger(DisruptorWithJGraphTExample.class.getName());
    
    // 定义事件类
    static class TaskEvent {
        private int taskId;

        public int getTaskId() {
            return taskId;
        }

        public void setTaskId(int taskId) {
            this.taskId = taskId;
        }
    }

    // 定义事件处理器
    static class TaskHandler implements com.lmax.disruptor.EventHandler<TaskEvent> {
        private final String name;

        public TaskHandler(String name) {
            this.name = name;
        }

        @Override
        public void onEvent(TaskEvent event, long sequence, boolean endOfBatch) {
            
            log.info(name + " processing task: " + event.getTaskId());
            // 模拟处理时间
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    public static void main(String[] args) throws InterruptedException {
        // 定义 Disruptor 的缓冲区大小
        int bufferSize = 1024;

        // 创建 Disruptor
        Disruptor<TaskEvent> disruptor = new Disruptor<>(
                TaskEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);

        // 创建 JGraphT 有向图
        Graph<String, DefaultEdge> graph = new DefaultDirectedGraph<>(DefaultEdge.class);

        // 添加顶点（每个顶点代表一个事件处理器）
        graph.addVertex("A");
        graph.addVertex("B");
        graph.addVertex("C");
        graph.addVertex("D");
        graph.addVertex("E");

        // 添加边（定义依赖关系）
        graph.addEdge("E", "B");
        graph.addEdge("A", "B"); // A -> B
        graph.addEdge("A", "C"); // A -> C
        graph.addEdge("B", "D"); // B -> D
        graph.addEdge("C", "D"); // C -> D

        // 检查图中是否有环
        CycleDetector<String, DefaultEdge> cycleDetector = new CycleDetector<>(graph);
        if (cycleDetector.detectCycles()) {
            throw new IllegalStateException("图中存在环，无法构建依赖关系！");
        }

        // 创建事件处理器实例
        Map<String, TaskHandler> handlers = new HashMap<>();
        handlers.put("A", new TaskHandler("HandlerA"));
        handlers.put("B", new TaskHandler("HandlerB"));
        handlers.put("C", new TaskHandler("HandlerC"));
        handlers.put("D", new TaskHandler("HandlerD"));
        handlers.put("E", new TaskHandler("e"));

        // 获取拓扑排序
        TopologicalOrderIterator<String, DefaultEdge> topologicalOrderIterator =
                new TopologicalOrderIterator<>(graph);

        // 动态设置 Disruptor 的事件处理器依赖关系
        List<String> topologicalOrder = new ArrayList<>();
        while (topologicalOrderIterator.hasNext()) {
            topologicalOrder.add(topologicalOrderIterator.next());
        }

        // 根据拓扑排序设置依赖关系
        for (int i = 0; i < topologicalOrder.size(); i++) {
            String currentHandlerName = topologicalOrder.get(i);
            TaskHandler currentHandler = handlers.get(currentHandlerName);

            // 获取当前顶点的前驱节点
            Set<DefaultEdge> incomingEdges = graph.incomingEdgesOf(currentHandlerName);
            if (incomingEdges.isEmpty()) {
                // 如果没有前驱节点，直接添加到 Disruptor
                disruptor.handleEventsWith(currentHandler);
            } else {
                // 如果有前驱节点，获取所有前驱节点的事件处理器
                List<TaskHandler> predecessors = new ArrayList<>();
                for (DefaultEdge edge : incomingEdges) {
                    String predecessorName = graph.getEdgeSource(edge);
                    predecessors.add(handlers.get(predecessorName));
                }
                // 设置依赖关系
                disruptor.after(predecessors.toArray(new TaskHandler[0])).handleEventsWith(currentHandler);
            }
        }

        // 启动 Disruptor
        RingBuffer<TaskEvent> ringBuffer = disruptor.start();

        // 发布事件
        for (int i = 0; i < 10; i++) {
            long sequence = ringBuffer.next();
            try {
                TaskEvent event = ringBuffer.get(sequence);
                event.setTaskId(i);
            } finally {
                ringBuffer.publish(sequence);
            }
            Thread.sleep(100); // 模拟事件发布的间隔
        }

        // 关闭 Disruptor
        disruptor.shutdown();
    }
}